package com.platform.apex.input;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.LinkedBlockingQueue;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.InputOperator;
import com.datatorrent.common.util.BaseOperator;

public class MysqlInputOperatorRealTime extends BaseOperator implements InputOperator {

	protected transient MyJdbcStore store;
	protected transient LinkedBlockingQueue emitQueue;
	public final transient DefaultOutputPort<Object> outputPort = new DefaultOutputPort<Object>();
	private String databaseUrl;
	private String userName;
	private String password;
	private int currentNumber;
	private String databaseDriver;
	private String tableName;

	private List<String> tableList;
	private Map<String, Integer> map = new HashMap<String, Integer>();

	@Override
	public void setup(OperatorContext context) {
		// TODO Auto-generated method stub
		super.setup(context);

		tableList = Arrays.asList(tableName.split(","));
		// 初始化查询进度
		for (String s : tableList) {
			map.put(s, 0);
		}

		store = new MyJdbcStore(databaseUrl, userName, password, databaseDriver);
		emitQueue = new LinkedBlockingQueue<>(10000);
	}

	@Override
	public void beginWindow(long windowId) {
		// TODO Auto-generated method stub
		super.beginWindow(windowId);

		// 监测数据表实时状况
		for (String s : tableList) {
			long queryCount = queryCount(s);
			int lastNumber = map.get(s);
			if (queryCount > map.get(s)) {
				int queryNumber = queryAndInsert(s, map.get(s));
				map.put(s, queryNumber+lastNumber);
			}
		}

	}

	@Override
	public void emitTuples() {
		// TODO Auto-generated method stub

		if(emitQueue.size()>0){
			try {
				outputPort.emit(emitQueue.take());
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	@Override
	public void endWindow() {
		// TODO Auto-generated method stub
		super.endWindow();
	}

	public long queryCount(String tableName) {
		Connection connSelect = null;
		PreparedStatement ps = null;
		ResultSet rs = null;

		String querySql = "Select count(1) as count from " + tableName;
		long count = 0;
		try {

			store.connect();
			connSelect = store.getConnection();

			ps = connSelect.prepareStatement(querySql);

			rs = ps.executeQuery();

			ResultSetMetaData rsmd = rs.getMetaData();
			// 可以得到有多少列
			int columnNum = rsmd.getColumnCount();
			while (rs.next()) {
				Object object = rs.getObject(1);
				count = (Long) object;
			}
			ps.close();
			connSelect.close();
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
		return count;
	}

	public int queryAndInsert(String tableName,int lastQueryNumber) {

		String querySql = "select * from "+tableName+" limit "+lastQueryNumber+", 10";
		Connection connSelect = null;
		PreparedStatement ps = null;
		ResultSet rs = null;

		Map<String, Object> mapToes = null;
		int number = 0;
		try {

			store.connect();
			connSelect = store.getConnection();

			ps = connSelect.prepareStatement(querySql);

			rs = ps.executeQuery();

			ResultSetMetaData rsmd = rs.getMetaData();
			// 可以得到有多少列
			int columnNum = rsmd.getColumnCount();
			
			while (rs.next()) {
				Object[] objects = new Object[columnNum];
				mapToes = new HashMap<String, Object>();
				mapToes.put("tableName", tableName);
				for (int i = 0; i < objects.length; i++) {
					objects[i] = rs.getObject(i + 1);
					String columnName = rsmd.getColumnName(i + 1);

					mapToes.put(columnName, objects[i]);
				}
				Map<String, Object> sortMapByKey = sortMapByKey(mapToes);
				emitQueue.put(sortMapByKey);
				number++;
			}
			ps.close();
			connSelect.close();
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}
		return number;
	}

	public static Map<String, Object> sortMapByKey(Map<String, Object> map) {
		if (map == null || map.isEmpty()) {
			return null;
		}

		Map<String, Object> sortMap = new TreeMap<String, Object>(new MapKeyComparator());

		sortMap.putAll(map);

		return sortMap;
	}

	public String getDatabaseUrl() {
		return databaseUrl;
	}

	public void setDatabaseUrl(String databaseUrl) {
		this.databaseUrl = databaseUrl;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public int getCurrentNumber() {
		return currentNumber;
	}

	public void setCurrentNumber(int currentNumber) {
		this.currentNumber = currentNumber;
	}

	public String getDatabaseDriver() {
		return databaseDriver;
	}

	public void setDatabaseDriver(String databaseDriver) {
		this.databaseDriver = databaseDriver;
	}

	public String getTableName() {
		return tableName;
	}

	public void setTableName(String tableName) {
		this.tableName = tableName;
	}

}
